package com.hadoop.chainmr;

import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
//import org.apache.hadoop.mapred.lib.ChainMapper;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.chain.ChainMapper;
import org.apache.hadoop.mapreduce.lib.chain.ChainReducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.mapreduce.lib.partition.HashPartitioner;


public class ChainMapReduce2 {

	private static final Path INPUT_PATH= new Path("hdfs://input/chainmr/*");
	private static final Path OUTPUT_PATH= new Path("hdfs://output/chainmr/*");
	private static final String OUTPUT_DIR= "hdfs://output/chainmr/*";
	
	public static void main(String[] args) throws IOException, URISyntaxException, ClassNotFoundException, InterruptedException {
		// TODO Auto-generated method stub
		
		Configuration conf = new Configuration();
		FileSystem fs = FileSystem.get(new URI(OUTPUT_DIR), conf);
		if(fs.exists(OUTPUT_PATH)) {
			fs.delete(OUTPUT_PATH, true);
			System.out.println("delete outputpath:" + OUTPUT_PATH.toString());
		}
		fs.close();
		
		Job job = Job.getInstance(conf, "chain mr2");
		job.setJarByClass(ChainMapReduce2.class);
		job.setInputFormatClass(TextInputFormat.class);
		
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(DoubleWritable.class);
		
		//设置分区
		job.setPartitionerClass(HashPartitioner.class);
		//设置reduce任务数量
		job.setNumReduceTasks(1);
		
		//设置输出的Key和Value类型
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(DoubleWritable.class);
		
		//设置最终输出的文件格式
		job.setOutputFormatClass(TextOutputFormat.class);
		FileInputFormat.addInputPath(job, INPUT_PATH);//设置输入数据文件路径
		FileOutputFormat.setOutputPath(job, OUTPUT_PATH);
		
		//设置chain mapper、reduce
		ChainMapper.addMapper(job, FilterMapper1.class, LongWritable.class, Text.class, Text.class, DoubleWritable.class, conf);
		
		ChainMapper.addMapper(job, FilterMapper2.class, Text.class, DoubleWritable.class, Text.class,
				DoubleWritable.class, conf);
		
		ChainReducer.setReducer(job, SumReducer3.class, Text.class, DoubleWritable.class, Text.class, DoubleWritable.class, conf);
		
		ChainMapper.addMapper(job, FilterMapper4.class, Text.class, DoubleWritable.class, Text.class,
				DoubleWritable.class, conf);
		
		//提交任务
		System.out.println(job.waitForCompletion(true));
	}

	//对数据进行抽取分离,过滤掉金额大于10,000的记录
	public static class FilterMapper1 extends Mapper<LongWritable, Text, Text, DoubleWritable>{
		
		//定义输出的key，value
		private Text outKey = new Text();
		private DoubleWritable outValue = new DoubleWritable();
		@Override
		protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
			String[] array = value.toString().split(" ");
			if(array.length == 2) {
				double va = Double.parseDouble(array[1]);
				if(va <= 10000) {
					outKey.set(array[0]);
					outValue.set(va);
					context.write(outKey, outValue);
				}
			}//if
		}
	}//FilterMapper1
	
	//在FilterMapper1的基础上，过滤掉金额小于100的记录
	public static class FilterMapper2 extends Mapper<Text, DoubleWritable, Text, DoubleWritable>{
		
		@Override
		protected void map(Text key, DoubleWritable value, Context context) throws IOException, InterruptedException {
			if(value.get() >= 100) {
				context.write(key, value);
			}
		}
	}
	
	//进行金额汇总
	public static class SumReducer3 extends Reducer<Text, DoubleWritable, Text, DoubleWritable>{
		
		private DoubleWritable outValue = new DoubleWritable();
		@Override
		protected void reduce(Text key, Iterable<DoubleWritable> values, Context context) throws IOException, InterruptedException {
			double sum =0;
			for(DoubleWritable value : values) {
				sum += value.get();
			}
			outValue.set(sum);
			context.write(key, outValue);
		}
	}
	
	//过滤掉商品名称大于8的商品
	public static class FilterMapper4 extends Mapper<Text, DoubleWritable, Text, DoubleWritable>{
		
		@Override
		protected void map(Text key, DoubleWritable value, Context context) throws IOException, InterruptedException {
			if(key.getLength() <= 8) {
				context.write(key, value);
			}
		}
	}
	
}
